package org.aksw.xoperator.sparql.http;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.regex.Pattern;

import org.aksw.xoperator.conf.AgentConfiguration;
import org.aksw.xoperator.sparql.AbstractThreadPooledSparqler;
import org.aksw.xoperator.sparql.SparqlEndpointIdentifier;
import org.aksw.xoperator.sparql.SparqlEndpointType;
import org.aksw.xoperator.sparql.SparqlExecutionFuture;
import org.aksw.xoperator.sparql.SparqlQuery;
import org.aksw.xoperator.sparql.SparqlResult;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class RemoteSparqlEndpoint extends AbstractThreadPooledSparqler {

	private AgentConfiguration conf;

	public RemoteSparqlEndpoint(ThreadPoolExecutor executor,
			AgentConfiguration conf) {
		super(executor);
		this.conf = conf;
		// TODO Auto-generated constructor stub
	}

	public void process(SparqlQuery query) {
		// TODO Auto-generated method stub

	}

	@Override
	public List<SparqlExecutionFuture<SparqlResult>> getFuturesFor(SparqlQuery query) {
		List<SparqlExecutionFuture<SparqlResult>> queries = new ArrayList<SparqlExecutionFuture<SparqlResult>>();
		// if query.getStoresToAsk is null, use all stores
		if (query.getStoresToAsk() == null) {
			for (RemoteSparqlEndpointConfiguration.SparqlEndpointConfigurationItem item : conf
					.getRemoteSparqlEndpoints().getEndpoints().values()) {
				queries.add(new SparqlExecutionFuture<SparqlResult>(
						new RemoteSparqlCallable(query, item),item.getEndpointUrl(),query.getQueryString()));
			}

		} else {
			// add the remote queries
			for (SparqlEndpointIdentifier ident : query.getStoresToAsk()) {
				// is the identifier of the URL-type? then create a temporary
				// configuration for it
				if (ident.getType().equals(SparqlEndpointType.URL)) {
					RemoteSparqlEndpointConfiguration.SparqlEndpointConfigurationItem tmpRemoteConf = new RemoteSparqlEndpointConfiguration.SparqlEndpointConfigurationItem(
							ident, ident.getName());
					queries.add(new SparqlExecutionFuture<SparqlResult>(
							new RemoteSparqlCallable(query, tmpRemoteConf),tmpRemoteConf.getEndpointUrl(),query.getQueryString()));
				} else {

					for (RemoteSparqlEndpointConfiguration.SparqlEndpointConfigurationItem item : conf
							.getRemoteSparqlEndpoints().getEndpoints().values()) {
						if (ident.getName().equals(item.getId().getName())) {
							queries.add(new SparqlExecutionFuture<SparqlResult>(
									new RemoteSparqlCallable(query, item),item.getEndpointUrl(),query.getQueryString()));
						}

					}
				}

			}
		}
		return queries;
	}

	class RemoteSparqlCallable implements Callable<SparqlResult> {

		private Log log = LogFactory.getLog(RemoteSparqlCallable.class);

		private SparqlQuery query;

		private RemoteSparqlEndpointConfiguration.SparqlEndpointConfigurationItem endpoint;

		public RemoteSparqlCallable(
				SparqlQuery query,
				RemoteSparqlEndpointConfiguration.SparqlEndpointConfigurationItem endpoint) {
			super();
			this.query = query;
			this.endpoint = endpoint;
		}

		public SparqlResult call() throws Exception {
			SparqlResult result = new SparqlResult();
			result.setAnsweredBy(endpoint.getId());
			query.addResult(result);

			HttpClient client = new HttpClient();
			HttpMethod get;
			try {

				get = new GetMethod(endpoint.getEndpointUrl());
				String queryParam = "query="
						+ java.net.URLEncoder.encode(query.getQueryString(),
								"UTF-8");
				if (get.getQueryString() != null
						&& !get.getQueryString().isEmpty()) {
					queryParam = get.getQueryString() + "&" + queryParam;
				}
				if (endpoint.getParameter() != null) {
					queryParam += endpoint.getParameter();
				}

				get.setQueryString(queryParam);

				this.log.info("Querying with: " + get.getURI());
				for (int i = 0; i < get.getRequestHeaders().length; i++) {
					log.info(get.getRequestHeaders()[i].toExternalForm());
				}
				try {
					client.executeMethod(get);
				} catch (IllegalArgumentException e) {
					log
							.error("Error in the datastore configuration, check datastore definition for store "
									+ endpoint.getId().getName());
				}
				final String res = get.getResponseBodyAsString();
				

				if (get.getStatusCode() == 200) {

					result.setAnswer(res);

				} else {

					// strip html from res

					Pattern pat = Pattern
							.compile("(<\\!--.*?-->)|(<\\!.*?>)|(<.+?>)",
									Pattern.DOTALL);
					String stripRes = pat.matcher(res).replaceAll("")
							.replaceAll("\\s+", " ");

					String message = "Store " + endpoint.getId().getName()
							+ " was unable to execute the query: " + stripRes;

					result.setErrorMsg(message);
					log.error(message);
				}

			} catch (UnsupportedEncodingException e1) {
				this.log.error(e1);
				result.setErrorMsg("Enconding Error while querying store : "
						+ endpoint.getId().getName());
			} catch (IOException e) {
				this.log.error(e);
				result.setErrorMsg("Error communicating with store "
						+ endpoint.getId().getName());

			}
			return result;
		}

	}

}
